經過 28 天的學習旅程,我們從 AWS AI 服務的基礎概念,一路實作到複雜的應用場景。
今天,我想整理這段時間累積的經驗,分享在使用 Amazon Bedrock 和 SageMaker 時的最佳實務,
希望能幫助大家少走彎路,更有效率地打造 AI 應用。
[這篇章主要是以總結為主]
選擇合適的服務
Bedrock vs SageMaker:何時使用哪個服務?
使用 Bedrock 的情境:
需要快速原型開發和驗證想法
應用場景符合基礎模型的能力範圍
團隊缺乏深度 ML 專業知識
需要多模型比較和切換的靈活性
重視開發速度和 Time-to-Market
使用 SageMaker 的情境:
需要高度客製化的模型
有特定領域的訓練資料
需要完整掌控模型的訓練過程
對模型性能有極致要求
需要處理敏感資料,無法使用第三方模型
實務建議:對於大多數企業應用,我建議採用「混合架構」:
用 Bedrock 處理通用 AI 任務(如對話、摘要、翻譯)
用 SageMaker 訓練專屬的業務模型(如欺詐檢測、需求預測)
透過 API Gateway 統一對外服務介面
考量面建立在幾個面向
成本優化 (參考過去文章)
Bedrock
SageMaker
安全實務
效能最佳化 (參考過去文章)
監控與可觀測性
開發流程實務 (參考過去文章)
例如 :
from typing import Dict, List, Optional
from pydantic import BaseModel, Field
class BedrockRequest(BaseModel):
    """Bedrock API 請求模型"""
    prompt: str = Field(..., description="使用者輸入的提示詞")
    model_id: str = Field(
        default="anthropic.claude-3-sonnet-20240229-v1:0",
        description="模型 ID"
    )
    max_tokens: int = Field(default=1024, ge=1, le=4096, description="最大 token 數")
    temperature: float = Field(default=1.0, ge=0, le=1, description="溫度參數")
    
    class Config:
        schema_extra = {
            "example": {
                "prompt": "請解釋什麼是機器學習",
                "model_id": "anthropic.claude-3-sonnet-20240229-v1:0",
                "max_tokens": 1024,
                "temperature": 0.7
            }
        }
class BedrockResponse(BaseModel):
    """Bedrock API 回應模型"""
    content: str = Field(..., description="生成的內容")
    model_id: str = Field(..., description="使用的模型 ID")
    tokens_used: int = Field(..., description="使用的 token 數量")
    latency_ms: float = Field(..., description="回應延遲(毫秒)")
例如 pytest
import pytest
from unittest.mock import Mock, patch
class TestBedrockIntegration:
    """Bedrock 整合測試"""
    
    @pytest.fixture
    def mock_bedrock_client(self):
        """Mock Bedrock 客戶端"""
        with patch('boto3.client') as mock_client:
            mock_instance = Mock()
            mock_client.return_value = mock_instance
            
            # 設定 mock 回應
            mock_instance.invoke_model.return_value = {
                'body': Mock(read=lambda: json.dumps({
                    'content': [{'text': '測試回應'}],
                    'usage': {'total_tokens': 100}
                }).encode())
            }
            
            yield mock_instance
    
    def test_invoke_bedrock_success(self, mock_bedrock_client):
        """測試成功調用 Bedrock"""
        result = invoke_bedrock("測試提示詞")
        assert result is not None
        assert 'content' in result
        
    def test_invoke_bedrock_with_cache(self, mock_bedrock_client):
        """測試快取機制"""
        prompt = "相同的提示詞"
        
        # 第一次調用
        result1 = invoke_bedrock_with_cache(prompt)
        
        # 第二次調用(應使用快取)
        result2 = invoke_bedrock_with_cache(prompt)
        
        # 驗證只調用一次 API
        assert mock_bedrock_client.invoke_model.call_count == 1
A/B testing
import random
class ABTestingFramework:
    def __init__(self):
        self.dynamodb = boto3.resource('dynamodb')
        self.results_table = self.dynamodb.Table('ab-test-results')
    
    def get_variant(self, user_id, experiment_name):
        """為使用者分配測試變體"""
        # 使用一致性雜湊確保同一使用者總是看到相同變體
        hash_value = hash(f"{user_id}{experiment_name}") % 100
        
        if hash_value < 50:
            return 'A'  # 控制組
        else:
            return 'B'  # 實驗組
    
    def log_result(self, user_id, experiment_name, variant, metric_name, value):
        """記錄測試結果"""
        self.results_table.put_item(Item={
            'experiment_id': f"{experiment_name}_{user_id}",
            'experiment_name': experiment_name,
            'variant': variant,
            'user_id': user_id,
            'metric_name': metric_name,
            'value': value,
            'timestamp': datetime.now().isoformat()
        })
# 使用範例:測試不同的 Prompt 策略
ab_test = ABTestingFramework()
def generate_response(user_id, user_query):
    variant = ab_test.get_variant(user_id, 'prompt_strategy_v1')
    
    if variant == 'A':
        # 控制組:基礎 Prompt
        prompt = f"請回答: {user_query}"
    else:
        # 實驗組:結構化 Prompt
        prompt = f"""作為專業助手,請依以下步驟回答:
1. 理解問題核心
2. 提供詳細解答
3. 給予實用建議
問題:{user_query}"""
    
    start_time = time.time()
    response = invoke_bedrock(prompt)
    latency = time.time() - start_time
    
    # 記錄指標
    ab_test.log_result(user_id, 'prompt_strategy_v1', variant, 'latency', latency)
    
    return response
And
明天見!